/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package cn.edu.thu.laud.objectstore.index;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.utils.ByteBufferUtil;

/**
 * Class that holds the key and the descriminator value if present
 * 
 * @author zhuoan
 *
 */
public class KVColumn implements Serializable{
	
	private byte[] rowKey;
	private Map<ByteArray, ByteArray> values;

	
	public KVColumn(ByteBuffer rowKey){
		this.rowKey = ByteBufferUtil.getArray(rowKey);
		values = new LinkedHashMap<ByteArray, ByteArray>();
	}
	
	public ByteBuffer getColumnValue(ByteBuffer key) {
		return values.get(ByteArray.getByteArray(key)).getByteBuffer();
	}
	
	public Map<ByteArray, ByteArray> getValues(){
		
		return values;
	}
	
	public ByteBuffer getRowKey(){
		
		return ByteBuffer.wrap(rowKey);
	}

	/**
	 * Add the result
	 * @param column
	 */
	public void addResult(Column column){
		values.put(ByteArray.getByteArray(column.name), ByteArray.getByteArray(column.value));
	}
	
	/* (non-Javadoc)
	 */
	public ByteBuffer getBytes() {
				
		try {
			
			ByteArrayOutputStream bos = new ByteArrayOutputStream();
			ObjectOutputStream oos = new ObjectOutputStream(bos);
			oos.writeObject(this);
			oos.flush();
			byte[] bytes = bos.toByteArray();
			oos.close();
			bos.close();			
			return ByteBuffer.wrap(bytes);
		} catch (Exception e) {
			//e.printStackTrace();
			throw new RuntimeException("Unable to serialize to object", e);
		}
	}

	/* (non-Javadoc)
	 */
	public static KVColumn createKVColumn(ByteBuffer buffer) {
		try {
			byte[] bs = ByteBufferUtil.getArray(buffer);
			ByteArrayInputStream bis = new ByteArrayInputStream(bs);
			ObjectInputStream ois;

			ois = new ObjectInputStream(bis);

			KVColumn serialized = (KVColumn) ois.readObject();
			ois.close();
			bis.close();

			return serialized;
		} catch (Exception e) {
			throw new RuntimeException("Unable to de-serialize to object", e);
		}
	}

	@Override
	public int hashCode() {
		
		final int prime = 31;
		int result = 1;
		result = prime * result + Arrays.hashCode(rowKey);
		result = prime * result + ((values == null) ? 0 : values.hashCode());
		return result;
	}

	@Override
	public boolean equals(Object obj) {
		
		if (this == obj) {
			return true;
		}
		if (obj == null) {
			return false;
		}
		if (!(obj instanceof KVColumn)) {
			return false;
		}
		KVColumn other = (KVColumn) obj;
		if (!Arrays.equals(rowKey, other.rowKey)) {
			return false;
		}
		if (values == null) {
			if (other.values != null) {
				return false;
			}
		} else if (!values.equals(other.values)) {
			return false;
		}
		return true;
	}

	@Override
	public String toString() {
		return "KVColumn [rowKey=" + new String(rowKey) + ", values="
				+ values + "]";
	}
	
	
	
	

}
